package com.bang.study.transaction;

import com.bang.study.constants.PublicConstants;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.ThreadFactoryImpl;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @ClassName: TransactionProducer
 * @Desc: TODO
 * @Auther: <Bang>
 * @Date: 2020/05/07 09:48
 * @Version: 1.0
 * @Modified By:
 */
public class TransactionProducer {

    public static void main(String[] args) throws Exception {
        TransactionMQProducer producer = new TransactionMQProducer(PublicConstants.TRANSACTION_GROUP);
        producer.setNamesrvAddr(PublicConstants.NAME_ADDR);
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 5, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(200), new ThreadFactoryImpl("TRANSACTION"));
        producer.setExecutorService(threadPoolExecutor);
        producer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                String transactionId = msg.getTransactionId();
                System.out.println("执行本地事务 transactionId=" + transactionId);
                if (StringUtils.equals("TAGA", msg.getTags())) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                } else if (StringUtils.equals("TAGB", msg.getTags())) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                } else {
                    return LocalTransactionState.UNKNOW;
                }
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.out.println("MQ检查消息Tag【"+msg.getTags()+"】的本地事务执行结果");
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });
        producer.start();
        String[] tagArr = {"TAGA", "TAGB", "TAGC"};
        for (int i = 0; i < 3; i++) {
            String TAG = tagArr[i];
            Message message = new Message(PublicConstants.TRANSACTION_TOPIC, TAG, "KEY" + i,
                    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            SendResult sendResult = producer.sendMessageInTransaction(message, null);
            System.out.printf("发送结果: %s%n", sendResult);
            Thread.sleep(1000);
        }

    }
}
